package day05

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 2.5	Action行动算子
 *
 * 行动算子是触发了整个作业的执行。因为转换算子都是懒加载，并不会立即执行。
 */
object ActionOperator {

}

/**
 * 2.5.1	reduce()聚合
 *
 * 函数签名：def reduce(f: (T, T) => T): T
 *
 * 功能说明：f函数聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据。
 */
object ActionOperator_reduce {
  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))

    // 调用reduce()
    val sum: Int = rdd.reduce(_ + _)

    println(sum)

    // 关闭连接
    sc.stop()

  }
}

/**
 * 2.5.2	collect()以数组的形式返回数据集
 *
 * 函数签名：def collect(): Array[T]
 *
 * 功能说明：在驱动程序中，以数组Array的形式返回数据集的所有元素。
 *
 * 注意：所有的数据都会被拉取到Driver端，慎用
 */
object ActionOperator_collect {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))

    // 调用collect()方法，收集数据到Driver端
    rdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.3	count()返回RDD中元素个数
 *
 * 函数签名：def count(): Long
 *
 * 功能说明：返回RDD中元素的个数
 */
object ActionOperator_count {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))

    // 调用count()，返回RDD中元素的个数
    val count: Long = rdd.count()

    println(count)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.4	first()返回RDD中的第一个元素
 *
 * 函数签名： def first(): T
 *
 * 功能说明：返回RDD中的第一个元素
 */
object ActionOperator_first {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))

    // 调用first()，返回RDD中的第一个元素
    val firstElement: Int = rdd.first()

    println(firstElement)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.5	take()返回由RDD前n个元素组成的数组
 *
 * 函数签名： def take(num: Int): Array[T]
 *
 * 功能说明：返回一个由RDD的前n个元素组成的数组
 */
object ActionOperator_take {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5))

    // 调用take()，返回前2个元素组成的数组
    val array: Array[Int] = rdd.take(num = 2)
    println(array.mkString(", "))

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.6	takeOrdered()返回该RDD排序后前n个元素组成的数组
 *
 * 函数签名： def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
 *
 * 功能说明：返回该RDD排序后的前n个元素组成的数组
 */
object ActionOperator_takeOrdered {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(2, 1, 3, 7, 5))

    // 调用takeOrdered()
    val thirdPlace: Array[Int] = rdd.takeOrdered(3)

    println(thirdPlace.mkString(", "))

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.7	aggregate()
 *
 * 函数签名：def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
 *
 * 功能说明：aggregate函数将每个分区的元素通过分区内逻辑和初始值进行聚合，然后用分区间逻辑和
 * 初始值(zeroValue)进行操作。
 *
 * 注意：分区间逻辑再次使用初始值和aggregateByKey是有区别的
 *
 */
object ActionOperator_aggregate {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 8)

    // 调用aggregate()，将所有元素相加得到结果
    val result1: Int = rdd.aggregate(0)(_ + _, _ + _)

    val result2: Int = rdd.aggregate(10)(_ + _, _ + _)

    println(result1)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.8	fold()算子
 *
 * 函数签名：def fold(zeroValue: T)(op: (T, T) => T): T
 *
 * 功能说明：折叠操作，aggregate的简化操作，即：分区内逻辑和分区间逻辑相同
 */
object ActionOperator_fold {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 8)

    // 调用fold()，初始值为10，分区内和分区间的计算规则相同
    val result: Int = rdd.fold(10)(_ + _)

    val result2: Int = rdd.fold(0)(_ + _)

    println(result2)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.9	countByKey()统计每种key的个数
 *
 * 函数签名：def countByKey(): Map[K, Long]
 *
 * 功能说明：统计每种key的个数
 */
object ActionOperator_countByKey {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    // 创建RDD
    val rdd: RDD[(Int, String)] = sc.makeRDD(List((1, "a"), (1, "a"), (1, "a"), (2, "b"), (3, "c"), (3, "c")))

    // 调用countByKey算子，统计每种key的个数
    val countRes: collection.Map[Int, Long] = rdd.countByKey()

    println(countRes)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.5.10	save相关算子
 *
 * 1）saveAsTextFile(path)保存成Text文件
 *
 * （1）函数签名
 *
 * （2）功能说明：将数据集的元素以textfile的形式保存到HDFS文件系统或者
 * 其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本
 *
 * 2）saveAsSequenceFile(path) 保存成Sequencefile文件
 * （1）函数签名
 *
 * （2）功能说明：将数据集中的元素以Hadoop Sequencefile的格式保存到指定的目录下，
 * 可以是HDFS或者其他Hadoop支持的文件系统。
 *
 * 注意：只有kv类型RDD有该操作，单值的没有
 *
 * 3）saveAsObjectFile(path) 序列化成对象保存到文件
 * （1）函数签名
 *
 * （2）功能说明：用于将RDD中的元素序列化成对象，存储到文件中。
 */
object ActionOperator_save {

  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
    rdd.saveAsTextFile("output/textFile")

    rdd.saveAsObjectFile("output/objectFile")

    val newRdd: RDD[(Int, Int)] = rdd.map((_, 1))

    newRdd.saveAsSequenceFile("output/sequenceFile")

    // 关闭连接
    sc.stop()
  }
}

object ActionOperator_foreach {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

    // 收集后打印
    rdd.collect().foreach(println)

    println("---------------------------------------")

    // 分布式打印
    rdd.foreach(println)

    // 关闭连接
    sc.stop()
  }
}


